/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream.source;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;

import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;

/**
 * Contacts the twitter sampling streaming api for iterative tweet generation.
 * Username and password must be provided in the properties, all else is handled
 * locally. Semantics for sampling: There is just one single iterator talking to
 * twitter. As a result concurrently contacting this iterator results in each
 * thread getting only a subset of all available tweets.
 */
public class TweetSamplingStream implements TweetSource, StatusListener {
  /** Logger for this class. */
  private static final Logger LOG = LoggerFactory.getLogger(TweetSamplingStream.class);
  /** Pointer to our single iterator. */
  private final TweetSamplingStreamIterator iterator = new TweetSamplingStreamIterator();
  /** Queue holding tweets we already retrieved. */
  private BlockingQueue<Status> tweetQueue;
  /** Set to true if retrieval fails or connection breaks down. */
  private final AtomicInteger failures = new AtomicInteger(0);

  private String streamConsumerkey;
  private String streamConsumersecret;
  private String streamAccesstoken;
  private String streamAccesstokensecret;

  private int streamTweetReceiveTimeout = 100;
  private int streamTweetTakeTimeout = 10;
  private int streamTweetQueueSize = 200;
  private int streamTweetErrorThreshold = 300;

 
  /**
   * Start sampling tweets from Twitter. Sampling will stop (with future items
   * being dropped) as soon as the internal blocking queue is full. Use the
   * iterator to take items out of that queue.
   * */  
  public void triggerHarvesting() {
    Preconditions.checkNotNull(streamConsumerkey, "streamConsumerkey missing!");
    Preconditions.checkNotNull(streamConsumersecret, "streamConsumersecret missing!");
    Preconditions.checkNotNull(streamAccesstoken, "streamAccesstoken missing!");
    Preconditions.checkNotNull(streamAccesstokensecret, "streamAccesstokensecret missing!");
    
    this.initQueue();
    
    ConfigurationBuilder cb = new ConfigurationBuilder();
    cb.setOAuthAccessToken(this.streamAccesstoken);
    cb.setOAuthAccessTokenSecret(this.streamAccesstokensecret);
    cb.setOAuthConsumerKey(this.streamConsumerkey);
    cb.setOAuthConsumerSecret(this.streamConsumersecret);

    TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
    twitterStream.addListener(this);
    twitterStream.sample();
  }
  
  void initQueue() {
    this.tweetQueue = new LinkedBlockingQueue<Status>(this.streamTweetQueueSize);    
  }

  /** {@inheritDoc} */
  @Override
  public Iterator<Status> iterator() {
    return this.iterator;
  }

  /**
   * Acts as connecting point between twitter4j and the twitter sink.
   * */
  private class TweetSamplingStreamIterator implements Iterator<Status> {
    /** {@inheritDoc} */
    @Override
    public boolean hasNext() {

      if (failures.get() >= streamTweetErrorThreshold) {
	LOG.error("Stopping twitter streaming. Too many errors: " + failures);
      }
      return (failures.get() < streamTweetErrorThreshold);
    }

    /** {@inheritDoc} */
    @Override
    public Status next() {
      try {
	return tweetQueue.poll(streamTweetTakeTimeout, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
	LOG.info("Failed to retrieve any tweets in " + streamTweetTakeTimeout + " seconds. Increasing failure counter by one.");
	failures.incrementAndGet();
      }
      return null;
    }

    /** Operation is not supported for this iterator. */
    @Override
    public void remove() {
      throw new UnsupportedOperationException("Remove on twitter streams not possible.");
    }
  }

  /** {@inheritDoc} */
  @Override
  public void onStatus(Status status) {
    try {

      if (!this.tweetQueue.offer(status, streamTweetReceiveTimeout, TimeUnit.SECONDS)) {
	LOG.warn("Failed to add current tweet content to queue: Dropping as queue seems full.");
      }
    } catch (InterruptedException e) {
      failures.incrementAndGet();
      LOG.warn("Could not process current tweet due to exception. Increasing failure count by one and continuing to process. "
	  + e.getMessage());
    }
  }

  /** {@inheritDoc} */
  @Override
  public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
    // ignoring
  }

  /** {@inheritDoc} */
  @Override
  public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
    // ignoring
  }

  /** {@inheritDoc} */
  @Override
  public void onException(Exception e) {
    failures.incrementAndGet();
  }

  @Override
  public void onScrubGeo(long arg0, long arg1) {
    // ignoring
  }

  /**
   * @param streamConsumerkey the streamConsumerkey to set
   */
  public void setStreamConsumerkey(String streamConsumerkey) {
    this.streamConsumerkey = streamConsumerkey;
  }

  /**
   * @param streamConsumersecret the streamConsumersecret to set
   */
  public void setStreamConsumersecret(String streamConsumersecret) {
    this.streamConsumersecret = streamConsumersecret;
  }

  /**
   * @param streamAccesstoken the streamAccesstoken to set
   */
  public void setStreamAccesstoken(String streamAccesstoken) {
    this.streamAccesstoken = streamAccesstoken;
  }

  /**
   * @param streamAccesstokensecret the streamAccesstokensecret to set
   */
  public void setStreamAccesstokensecret(String streamAccesstokensecret) {
    this.streamAccesstokensecret = streamAccesstokensecret;
  }

  /**
   * @param streamTweetReceiveTimeout the streamTweetReceiveTimeout to set
   */
  public void setStreamTweetReceiveTimeout(int streamTweetReceiveTimeout) {
    this.streamTweetReceiveTimeout = streamTweetReceiveTimeout;
  }

  /**
   * @param streamTweetTakeTimeout the streamTweetTakeTimeout to set
   */
  public void setStreamTweetTakeTimeout(int streamTweetTakeTimeout) {
    this.streamTweetTakeTimeout = streamTweetTakeTimeout;
  }

  /**
   * @param streamTweetQueueSize the streamTweetQueueSize to set
   */
  public void setStreamTweetQueueSize(int streamTweetQueueSize) {
    this.streamTweetQueueSize = streamTweetQueueSize;
  }

  /**
   * @param streamTweetErrorThreshold the streamTweetErrorThreshold to set
   */
  public void setStreamTweetErrorThreshold(int streamTweetErrorThreshold) {
    this.streamTweetErrorThreshold = streamTweetErrorThreshold;
  }
}
